package com.demo.java.disruptor.netty.common;

import com.demo.java.disruptor.netty.client.ClientConsumer;
import com.demo.java.disruptor.netty.server.ServerConsumer;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
import lombok.Data;

import java.util.concurrent.Executors;

public class DisruptorFactory {

    private static class DisruptorFactoryHolder {
        private static DisruptorFactory instance = new DisruptorFactory();
    }
    public static DisruptorFactory getInstance() {
        return DisruptorFactoryHolder.instance;
    }

    private int consumerSize = 10;
    private RingBuffer<TranslatorDataWrap> ringBuffer;
    private SequenceBarrier sequenceBarrier;
    private WorkerPool<TranslatorDataWrap> workerPool;
    public void init(ProducerType producerType, int size, WaitStrategy waitStrategy, WorkHandler<TranslatorDataWrap>[] consumers) {
        EventFactory<TranslatorDataWrap> eventFactory = TranslatorDataWrap::new;
        ExceptionHandler<TranslatorDataWrap> exceptionHandler = new ExceptionHandlerImpl();
        this.ringBuffer = RingBuffer.create(producerType, eventFactory, size, waitStrategy);
        this.sequenceBarrier = this.ringBuffer.newBarrier();
        this.workerPool = new WorkerPool<>(this.ringBuffer, this.sequenceBarrier, exceptionHandler, consumers);
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2));
    }

    private static class ExceptionHandlerImpl implements ExceptionHandler<TranslatorDataWrap> {
        public void handleEventException(Throwable throwable, long l, TranslatorDataWrap translatorDataWrap) {
            System.out.println("handle event exception");
            throwable.printStackTrace();
        }
        public void handleOnStartException(Throwable throwable) {
            System.out.println("handle on start exception");
            throwable.printStackTrace();
        }
        public void handleOnShutdownException(Throwable throwable) {
            System.out.println("handle on shutdown exception");
            throwable.printStackTrace();
        }
    }

    private RingBuffer<TranslatorDataWrap> getRingBuffer() {
        return this.ringBuffer;
    }

    public static AbstractConsumer createClientConsumer() {
        return new ClientConsumer();
    }
    public static AbstractConsumer createServerConsumer() {
        return new ServerConsumer();
    }
    public static Producer createProducer() {
        return new Producer(getInstance().getRingBuffer());
    }

    @Data
    public static class Producer {
        private RingBuffer<TranslatorDataWrap> ringBuffer;
        public Producer(RingBuffer<TranslatorDataWrap> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }

        public void sendData(ChannelHandlerContext ctx, TranslatorData data, boolean release) {
            long sequence = this.ringBuffer.next();
            TranslatorDataWrap tmp = this.ringBuffer.get(sequence);
            tmp.setData(data);
            tmp.setCtx(ctx);
            this.ringBuffer.publish(sequence);
            if (release) {
                ReferenceCountUtil.release(data);
            }
        }
    }
}
